
package com.flybear.aqi.server;

import com.flybear.aqi.client.ElasticSearchClient;
import com.flybear.aqi.client.ElasticSearchClientFactory;
import com.flybear.aqi.client.EsUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.Delimiters;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.SelfSignedCertificate;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * Echoes back any received data from a client.
 */
public final class AgentServer {
    public static final Logger logger = LoggerFactory.getLogger(AgentServer.class);
    static final int SIZE = Integer.parseInt(System.getProperty("size", "1040"));
    static final boolean SSL = System.getProperty("ssl") != null;
    static final int PORT = Integer.parseInt(System.getProperty("port", "8007"));

    public static void main(String[] args) throws Exception {
        // Configure SSL.
        final SslContext sslCtx;
        if (SSL) {
            SelfSignedCertificate ssc = new SelfSignedCertificate();
            sslCtx = SslContextBuilder.forServer(ssc.certificate(), ssc.privateKey()).build();
        } else {
            sslCtx = null;
        }

        // Configure the server.
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        ElasticSearchClient client=null;
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .option(ChannelOption.SO_BACKLOG, 2048)
                    .option(ChannelOption.TCP_NODELAY, true)
//                    .option(ChannelOption.SO_RCVBUF,32)
//                    .option(ChannelOption.SO_SNDBUF,32)
//                    .option(ChannelOption.RCVBUF_ALLOCATOR, AdaptiveRecvByteBufAllocator.DEFAULT)
//                    .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)//ReferenceCountUtil.release(msg)
//             .handler(new LoggingHandler(LogLevel.INFO))
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     if (sslCtx != null) {
                         p.addLast(sslCtx.newHandler(ch.alloc()));
                     }
//                     p.addLast("idleStateHandler", new IdleStateHandler(0, 0, 180));
                     //p.addLast(new LoggingHandler(LogLevel.INFO));
                     // 解码器
                    // 基于换行符号
//                     p.addLast(new LineBasedFrameDecoder( SIZE));

                     // 基于指定字符串【换行符，这样功能等同于LineBasedFrameDecoder】
//		e.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, false, Delimiters.lineDelimiter()));

                     // 基于最大长度
//		e.pipeline().addLast(new FixedLengthFrameDecoder(4));

                     // 解码转Sring
//                     p.addLast(new StringDecoder());
                     p.addLast(new DelimiterBasedFrameDecoder(8192, Delimiters.lineDelimiter()));
                     p.addLast(new CustomerDecoder(CharsetUtil.UTF_8));
                     p.addLast(new StringEncoder(CharsetUtil.UTF_8));


                     p.addLast(new AgentServerHandler());
                 }
             });

            // Start the server.
            ChannelFuture f = b.bind(PORT).sync();

            try {
                ElasticSearchClientFactory clientFactory = new ElasticSearchClientFactory();
                String clusterName = EsUtil.getkey("name");
                String[] serverAddresses = StringUtils.deleteWhitespace(
                        EsUtil.getkey("ip")).split(",");
                logger.info("ElasticSearch client " + clusterName + " started");
                client = clientFactory.getClient(ElasticSearchClientFactory.HighLevelRestClient, serverAddresses,
                        clusterName, null, null);
                Map<String, Object> conf = new HashMap<String, Object>();
//      conf.put("dbex",this.db);
                client.configure(new HashMap<String, String>((Map) EsUtil.getproperties()), conf);


            } catch (Exception ex) {
                ex.printStackTrace();

                if (client != null) {
                    client.close();
                }
            }
            EsUtil.client=client;
            // Wait until the server socket is closed.
            f.channel().closeFuture().sync();
        } finally {
            // Shut down all event loops to terminate all threads.
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            if (client != null) {
                client.close();
            }
        }
    }
}
